package com.example.iot.service.websocket.impl;

import com.example.iot.entity.dto.TsData;
import com.example.iot.entity.id.EntityIdFactory;
import com.example.iot.entity.id.EntityType;
import com.example.iot.service.ts.TimeSeriesService;
import com.example.iot.service.websocket.SubscriptionService;
import com.example.iot.service.websocket.WebSocketService;
import com.example.iot.ts.TimeSeriesSubscriptionInfo;
import com.example.iot.websocket.SessionEvent;
import com.example.iot.websocket.WebSocketMsgEndpoint;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.socket.WebSocketSession;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author ChenXiangpeng
 */
@Service
@Slf4j
public class WebSocketServiceImpl implements WebSocketService {

    @Autowired
    private WebSocketMsgEndpoint webSocketMsgEndpoint;
    @Autowired
    private SubscriptionService subscriptionService;
    @Autowired
    private TimeSeriesService timeSeriesService;

    private static final ObjectMapper jsonMapper = new ObjectMapper();
    private ExecutorService executor;

    @PostConstruct
    public void initExecutor() {
        executor = Executors.newWorkStealingPool(50);
    }

    @PreDestroy
    public void shutdownExecutor() {
        if (executor != null) {
            executor.shutdownNow();
        }
    }

    @Override
    public void handleWebSocketMsg(WebSocketSession session, String message) {
        try{
            JsonNode jsonNode = jsonMapper.readTree(message);
            String entityId = jsonNode.get("entityId").asText();
            String queryType = jsonNode.get("queryType").asText();
            String sessionId = session.getId();
            if ("getAll".equals(queryType)){
                handleWsTimeSeriesSubscription(sessionId, entityId, new FutureCallback<List<TsData>>() {
                    @SneakyThrows
                    @Override
                    public void onSuccess(@Nullable List<TsData> result) {
                        sendMsg(sessionId, jsonMapper.writeValueAsString(result));
                    }

                    @Override
                    public void onFailure(Throwable t) {
                        log.error("错误{}", t.getMessage());
                        sendMsg(sessionId, t.getMessage());
                    }
                });
            }else{
                handleWsTimeSeriesSubscription(sessionId, entityId);
            }

        }catch (JsonProcessingException jsonProcessingException){
            log.info("json消息反序列化失败, [{}]", jsonProcessingException.getMessage());
            // todo 返回错误信息TimeSeries
        }

    }

    @Override
    public void handleWebSocketSessionEvent(WebSocketSession session, SessionEvent sessionEvent) {
        String sessionId = session.getId();
        switch (sessionEvent){
            case ESTABLISHED:
                break;
            case ERROR:
                log.error("websocket未知错误");
                break;
            case CLOSED:
                subscriptionService.cancelAllSubscription(sessionId);
                break;
            default:
                break;
        }
    }

    private void handleWsTimeSeriesSubscription(String sessionId, String entityId){
        // 异步执行 注册消息订阅
        CompletableFuture.runAsync(()->{
            TimeSeriesSubscriptionInfo<String> subscriptionInfo = new TimeSeriesSubscriptionInfo<>(sessionId,
                            EntityIdFactory.getByTypeAndUuid(EntityType.DEVICE, entityId),
                            this::sendMsg);
            subscriptionService.addSubscription(subscriptionInfo);
        }, executor);
    }

    private void handleWsTimeSeriesSubscription(String sessionId, String entityId, FutureCallback<List<TsData>> callback){
        // 异步执行查询最新数据
        Futures.addCallback(timeSeriesService.getAllLatest(entityId, executor), callback, executor);
        // 异步执行 注册消息订阅
        CompletableFuture.runAsync(() -> {
            TimeSeriesSubscriptionInfo<String> subscriptionInfo = new TimeSeriesSubscriptionInfo<>(sessionId,
                    EntityIdFactory.getByTypeAndUuid(EntityType.DEVICE, entityId),
                    this::sendMsg);
            subscriptionService.addSubscription(subscriptionInfo);
        }, executor);
    }

    private void sendMsg(String sessionId, String message){
        // 根据sessionId发送消息
        executor.submit(() -> webSocketMsgEndpoint.send(sessionId, message));
    }
}
